package com.iyang.boothbaseapitest.utils;

import com.iyang.boothbaseapitest.config.HBaseConn;
import org.apache.hadoop.hbase.HColumnDescriptor;
import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.filter.FilterList;
import org.apache.hadoop.hbase.util.Bytes;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Arrays;
import java.util.List;

/**
 * @Author: Mu_YI
 * @Date: 2019/3/13 22:20
 * @Version 1.0
 */
public class HBaseUtils {

    private static final Logger LOGGER = LoggerFactory.getLogger(HBaseUtils.class);

    /**
     *
     * @param tableName  表名
     * @param cfs        列族的数组
     * @return           创建是否成功
     */
    public static boolean createTable(String tableName,String [] cfs){
        try(HBaseAdmin admin = (HBaseAdmin) HBaseConn.getHBaseConn().getAdmin()){
            if(admin.tableExists(tableName)){
                return false;
            }
            HTableDescriptor tableDescriptor = new HTableDescriptor(TableName.valueOf(tableName));
            Arrays.stream(cfs).forEach(cf->{
                HColumnDescriptor columnDescriptor = new HColumnDescriptor(cf);
                columnDescriptor.setMaxVersions(1);
                tableDescriptor.addFamily(columnDescriptor);
            });
            admin.createTable(tableDescriptor);
        }catch (Exception e){
            LOGGER.error(e.getMessage(),e);
        }
        return true;
    }

    public static boolean deleteTable(String tableName){
        try(HBaseAdmin hBaseAdmin = (HBaseAdmin)HBaseConn.getHBaseConn().getAdmin()){
            hBaseAdmin.disableTable(tableName);
            hBaseAdmin.deleteTable(tableName);
        }catch (Exception e){
            LOGGER.error(e.getMessage(),e);
        }
        return true;
    }

    public static boolean putRow(String tableName,String rowKey,String cfName,String qualifier,String data){
        try(Table table = HBaseConn.getTable(tableName)){
            Put put = new Put(Bytes.toBytes(rowKey));
            put.addColumn(Bytes.toBytes(cfName),Bytes.toBytes(qualifier),Bytes.toBytes(data));
            table.put(put);
        }catch (Exception e){
            LOGGER.error(e.getMessage(),e);
        }
        return true;
    }


    public static boolean putRows(String tableName, List<Put> puts){
        try(Table table = HBaseConn.getTable(tableName)){
            table.put(puts);
        }catch (Exception e){
            LOGGER.error(e.getMessage(),e);
        }
        return true;
    }

    public static Result getRow(String tableName,String rowKey){
        try(Table table = HBaseConn.getTable(tableName)){
            Get get = new Get(Bytes.toBytes(rowKey));
            return  table.get(get);
        }catch (Exception e){
            LOGGER.error(e.getMessage(),e);
        }
        return null;
    }

    public static Result getRow(String tableName, String rowKey, FilterList filterList){
        try(Table table = HBaseConn.getTable(tableName)){
            Get get = new Get(Bytes.toBytes(rowKey));
            get.setFilter(filterList);
            return  table.get(get);
        }catch (Exception e){
            LOGGER.error(e.getMessage(),e);
        }
        return null;
    }

    public static ResultScanner getScanner(String tableName){
        try(Table table = HBaseConn.getTable(tableName)){
           Scan scan = new Scan();
           scan.setCaching(1000);
           return table.getScanner(scan);
        }catch (Exception e){
            LOGGER.error(e.getMessage(),e);
        }
        return null;
    }

    public static ResultScanner getScanner(String tableName,String startRowKey,String endRowKey){
        try(Table table = HBaseConn.getTable(tableName)){
            Scan scan = new Scan();
            scan.setStartRow(Bytes.toBytes(startRowKey));
            scan.setStopRow(Bytes.toBytes(endRowKey));
            scan.setCaching(1000);
            return table.getScanner(scan);
        }catch (Exception e){
            LOGGER.error(e.getMessage(),e);
        }
        return null;
    }

    public static ResultScanner getScanner(String tableName,String startRowKey,String endRowKey,FilterList filterList){
        try(Table table = HBaseConn.getTable(tableName)){
            Scan scan = new Scan();
            scan.setStartRow(Bytes.toBytes(startRowKey));
            scan.setStopRow(Bytes.toBytes(endRowKey));
            scan.setFilter(filterList);
            scan.setCaching(1000);
            return table.getScanner(scan);
        }catch (Exception e){
            LOGGER.error(e.getMessage(),e);
        }
        return null;
    }

    /**
     * 删除
     * @param tableName  表名
     * @param rowKey     唯一表示行
     * @return
     */
    public static boolean deleteRow(String tableName,String rowKey){
        try(Table table = HBaseConn.getTable(tableName)){
            Delete delete = new Delete(Bytes.toBytes(rowKey));
            table.delete(delete);
        }catch (Exception e){
            LOGGER.error(e.getMessage(),e);
        }
        return true;
    }


    public static boolean deleteColumnFamily(String tableName,String cfName){
        try(HBaseAdmin hBaseAdmin = (HBaseAdmin)HBaseConn.getHBaseConn().getAdmin()){
           hBaseAdmin.deleteColumn(tableName,cfName);
        }catch (Exception e){
            LOGGER.error(e.getMessage(),e);
        }
        return true;
    }

    public static boolean deleteQualifer(String tableName,String keyRow,String cfName,String qualifer){
        try(Table table = HBaseConn.getTable(tableName)){
           Delete delete = new Delete(Bytes.toBytes(keyRow));
           delete.addColumn(Bytes.toBytes(cfName),Bytes.toBytes(qualifer));
           table.delete(delete);
        }catch (Exception e){
            LOGGER.error(e.getMessage(),e);
        }
        return true;
    }
}
